package com.shujia.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object DSLApiDemo2 {
  def main(args: Array[String]): Unit = {
    val ss: SparkSession = SparkSession.builder()
      .config("spark.sql.shuffle.partitions","1")
      .master("local")
      .appName("DSL语法学习")
      .getOrCreate()

    val sc: SparkContext = ss.sparkContext

    //----------------------------------------------------

    //读取json数据
    //老版读取json数据方式，先指定数据格式，再指定路径
    //    val df1: DataFrame = ss.read
    //      .format("json")
    //      .load("spark/data/students.json")

    //新版直接使用json方法读取数据
//    val df1: DataFrame = ss.read.json("spark/data/students.json")
    //默认只会展示前20行数据，不会完整显示一行所有内容
    //    df1.show(false)
    //自定义展示的行数
    //    df1.show(100)
    //完整展示前30条数据
    //    df1.show(30, truncate = false)

    //查看DataFrame字段结构
    //    df1.printSchema()

    import org.apache.spark.sql.functions._
    import ss.implicits._
    //------------------------DSL语法----------------------------
    /**
     * select
     */
    //查看姓名和年龄
    //    df1.select("name","age").show(false)
    //    df1.select($"name",$"age").show(false)

    //查看姓名和年龄,同时将年龄+1
    //    df1.select("name","age"+1).show() // 不加$，取源DF中普通的列名"age"+1意味着要取一个名字叫"age1"的列
    //    df1.select($"name",$"age"+1).show() // 加了$，将对应的列名变成一个列对象，将列值取出做操作

    //使用selectExpr查询，双引号中的内容看作是一个表达式
    //    df1.selectExpr("name","age+1 as new_age").show()
    //    df1.select($"name",$"age"+1 as "new_age").show()

    /**
     * where
     */
    //需求：查询性别是女生，且班级是理科二班的
    //    df1.where("gender='女'").show()
    // === 等同于sql中 =
    //    df1.where($"gender" === "女").show()
    // =!= 等同于sql中 != 或 <>
    //    df1.where($"gender" =!= "女").show()

    //    df1.where("gender='女' and clazz='理科二班'").show()
    //    df1.where($"gender" === "女" and $"clazz" === "理科二班").show()

    /**
     * group by
     *
     * 分组，特点：非分组字段无法出现最终select中
     */
    //    df1.groupBy("clazz")
    //      .agg(count(lit(1)) as "clazzNum")
    //      .show()

    /**
     * order by
     *
     */
//    df1.groupBy("clazz")
//      .agg(count(lit(1)) as "clazzNum")
//      .orderBy($"clazzNum".desc)
//      .show()



    /**
     * join
     */
    val studentDF: DataFrame = ss.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
      .load("spark/data/students.txt")

    val scoreDF: DataFrame = ss.read
      .format("csv")
      .option("sep", ",")
      .schema("sid STRING,subject_id STRING,score INT")
      .load("spark/data/score.txt")

    //当关联的字段名不一样的时候，可以指定连接的方式
//    studentDF.join(scoreDF,$"id" === $"sid", "left")
//      .show()

    //当关联的字段名一样的时候
    //若单纯传入一个字符串名的时候，无法指定连接的方式
//    studentDF.join(scoreDF,"id")
//      .show()
    //若关联的字段名一样的时候，且也想指定其他的连接方式, 将字段变成Seq类型
//    studentDF.join(scoreDF,Seq("id"),"left")
//      .show()


    /**
     * 开窗：开窗不会改变原有的数据条数，只会新增一列
     *
     * 聚合开窗函数
     * 排序开窗函数
     *
     */
    //需求：求出每个班级总分前3的学生详细信息
    //读取学生数据
    //纯sql的方式分析
//    studentDF.createOrReplaceTempView("students")
//    scoreDF.createOrReplaceTempView("scores")
//
//    ss.sql(
//      """
//        |
//        |select
//        |tt1.id as id,
//        |tt1.sumScore as sumScore,
//        |tt1.name as name,
//        |tt1.age as age,
//        |tt1.clazz as clazz,
//        |tt1.sumScore as sumScore
//        |from
//        |(
//        |select
//        |t1.id as id,
//        |t2.name as name,
//        |t2.age as age,
//        |t2.clazz,
//        |t1.sumScore as sumScore,
//        |row_number() over(partition by t2.clazz order by t1.sumScore desc) as rn
//        |from
//        |(
//        |select
//        |id,
//        |sum(score) as sumScore
//        |from
//        |scores
//        |group by id) t1
//        |join
//        |students t2
//        |on (t1.id=t2.id)) tt1
//        |where tt1.rn <= 3
//        |
//        |""".stripMargin).show()

    // DSL语法分析
    scoreDF.groupBy("sid")
      .agg(sum("score") as "sumScore")
      .join(studentDF,$"sid" === $"id")
      .select($"id",$"name",$"age",$"gender",$"clazz",$"sumScore",row_number() over Window.partitionBy($"clazz").orderBy($"sumScore".desc) as "rn")
      .where("rn <= 3")
      .show()




  }
}
